package com.lcf.user.controller;

import com.lcf.user.mapper.PortraitLibMapper;
import io.swagger.annotations.Api;
import io.swagger.annotations.ApiOperation;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.elasticsearch.client.transport.TransportClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.List;

@RestController
@RequestMapping("/sensetime/cluster")
@Api(value = "人像库服务", tags = {"ES库的操作"})
public class PortraitController {
    @Autowired
    private TransportClient client;

    @Autowired
    private PortraitLibMapper portraitLibMapper;

    @PostMapping("/portrait/add")
    @ApiOperation(value = "人像入库", notes = "人像库添加数据", response = Integer.class)
    public void portraitAdd() {

    }


    @PostMapping("/portrait/search")
    @ApiOperation(value = "获取聚类信息", notes = "获取聚类信息", response = Integer.class)
    public void portraitSearch() {

    }

    /**
     * data ： 对于data值的类型其实并没有限定，根据KafkaTemplate所定义的类型来决定。 data为List集合的则是用作批量消费。
     * ConsumerRecord：具体消费数据类，包含Headers信息、分区信息、时间戳等
     * Acknowledgment：用作Ack机制的接口
     * Consumer：消费者类，使用该类我们可以手动提交偏移量、控制消费速率等功能
     * containerFactory = "kafkaListenerContainerFactory"
     */
    @KafkaListener(groupId = "image2", topics = ("image-topic"), containerFactory = "kafkaListenerContainerFactory")
    public void ackTest(List<ConsumerRecord> records, Acknowledgment ack) {
        try {

            for (ConsumerRecord record : records) {
                Thread.sleep(100000);
                System.out.println(String.format("image-topic=======>partition = %d, offset = %d, key = %s, value = %s%n \n", record.partition(), record.offset(), record.key(), record.value()));
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            ack.acknowledge();//手动提交偏移量

        }
    }
}